package spring.cloud.tasks.executor.listener.failover;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import spring.cloud.tasks.executor.common.Service;
import spring.cloud.tasks.executor.listener.config.ConfigurationNode;
import spring.cloud.tasks.executor.node.TaskNodePaths;
import spring.cloud.tasks.executor.task.ExecuteNode;
import spring.cloud.tasks.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.utils.log.LogUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 作业失效转移服务.
 */
@Slf4j
@Setter
@Getter
public class FailoverService extends Service {
    public FailoverService(final TaskExecutor taskExecutor) {
        super(taskExecutor);
    }

    @Override
    public void start() {
    }

    /**
     * 设置失效的分片项标记.
     *
     * @param taskItemIndex 崩溃的作业项
     */
    public void createFailoverFlag(final int taskItemIndex) {
        if (!isFailoverFlagExisted(taskItemIndex)) {
            try {
                this.getTaskNodeStorage().getCuratorFramework().create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(TaskNodePaths.getTaskNodePath(taskId, FailoverNode.getItemsNode(taskItemIndex)));
                LogUtils.info(log, taskId, "{} - {} create failover flag of item {}", executorId, taskId, taskItemIndex);
            } catch (KeeperException.NodeExistsException e) { // NOSONAR
                LogUtils.debug(log, taskId, "{} - {} create failover flag of item {} failed, because it is already existing", executorId, taskId, taskItemIndex);
            } catch (Exception e) {
                LogUtils.error(log, taskId, e.getMessage(), e);
            }
        }
    }

    public boolean isFailoverFlagExisted(final Integer taskItemIndex) {
        return this.getTaskNodeStorage().isTaskNodeExisted(FailoverNode.getExecutionFailoverNode(taskItemIndex));
    }

    /**
     * 如果需要失效转移, 则设置作业失效转移.
     */
    public void failoverIfNecessary() {
        if (!needFailover()) {
            return;
        }
        this.getTaskNodeStorage().executeInLeader(FailoverNode.LATCH, new FailoverLeaderExecuteCallback(this), 1, TimeUnit.MINUTES, new FailoverTimeoutLeaderExecuteCallback(this));
    }

    boolean needFailover() {
        return this.getTaskNodeStorage().isTaskNodeExisted(FailoverNode.ITEMS_ROOT) && !this.getTaskNodeStorage().getTaskNodeChildrenKeys(FailoverNode.ITEMS_ROOT).isEmpty() && this.getTaskNodeStorage().isTaskNodeExisted(ConfigurationNode.ENABLED) && Boolean.parseBoolean(this.getTaskNodeStorage().getTaskNodeData(ConfigurationNode.ENABLED));
    }


    /**
     * 获取运行在本作业服务器的失效转移序列号.
     *
     * @return 运行在本作业服务器的失效转移序列号
     */
    public List<Integer> getFailoverTaskItemIndexList() {
        List<String> taskItemIndexStringList = this.getTaskNodeStorage().getTaskNodeChildrenKeys(ExecuteNode.ROOT);
        List<Integer> taskItemIndexList = new ArrayList<>(taskItemIndexStringList.size());
        for (String taskItemIndexString : taskItemIndexStringList) {
            int taskItemIndex = Integer.parseInt(taskItemIndexString);
            String executionFailoverNode = FailoverNode.getExecutionFailoverNode(taskItemIndex);
            if (this.getTaskNodeStorage().isTaskNodeExisted(executionFailoverNode) && executorId.equals(this.getTaskNodeStorage().getTaskNodeData(executionFailoverNode))) {
                taskItemIndexList.add(taskItemIndex);
            }
        }
        Collections.sort(taskItemIndexList);
        return taskItemIndexList;
    }

    /**
     * 获取运行在本作业服务器的被失效转移的序列号.
     *
     * @return 运行在本作业服务器的被失效转移的序列号
     */
    public List<Integer> getTakeOffTaskItemIndexList() {
        List<Integer> shardingTaskItemIndexList = taskExecutor.getShardingService().getShardingTaskItemIndexList();
        List<Integer> taskItemIndexList = new ArrayList<>(shardingTaskItemIndexList.size());
        for (int shardingTaskItemIndex : shardingTaskItemIndexList) {
            if (this.getTaskNodeStorage().isTaskNodeExisted(FailoverNode.getExecutionFailoverNode(shardingTaskItemIndex))) {
                taskItemIndexList.add(shardingTaskItemIndex);
            }
        }
        return taskItemIndexList;
    }

    /**
     * 删除作业失效转移信息.
     */
    public void removeFailoverFlag() {
        this.getTaskNodeStorage().removeTaskNodeIfExisted(FailoverNode.ITEMS_ROOT);
        for (String taskItemIndex : this.getTaskNodeStorage().getTaskNodeChildrenKeys(ExecuteNode.ROOT)) {
            this.getTaskNodeStorage().removeTaskNodeIfExisted(FailoverNode.getExecutionFailoverNode(Integer.parseInt(taskItemIndex)));
        }
    }

    /**
     * 更新执行完毕失效转移的分片项状态.
     *
     * @param taskItemIndex 执行完毕失效转移的分片项列表
     */
    public void updateFailoverComplete(final Integer taskItemIndex) {
        this.getTaskNodeStorage().removeTaskNodeIfExisted(FailoverNode.getExecutionFailoverNode(taskItemIndex));
    }


}
